package com.example.channel

import kotlinx.coroutines.GlobalScope
import kotlinx.coroutines.channels.BroadcastChannel
import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.channels.ReceiveChannel
import kotlinx.coroutines.channels.SendChannel
import kotlinx.coroutines.channels.actor
import kotlinx.coroutines.channels.broadcast
import kotlinx.coroutines.channels.produce
import kotlinx.coroutines.delay
import kotlinx.coroutines.joinAll
import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking
import org.junit.Test

class CoroutineTest01 {

    //通道-多路复用-安全并发


    @Test
    fun `test know channel`() = runBlocking<Unit> {
        val channel = Channel<Int>()
        val producer = GlobalScope.launch {
            var i = 0
            while (true) {
                delay(1000)
                channel.send(++i)
                println("send $i")
            }
        }

        val consumer = GlobalScope.launch {
            while (true) {
                delay(2000)
                val element = channel.receive()
                println("receive $element")
            }
        }

        joinAll(producer, consumer)
    }


    @Test
    fun `test iterate channel`() = runBlocking<Unit> {
        val channel = Channel<Int>(Channel.UNLIMITED)
        val producer = GlobalScope.launch {
            for (x in 1..5) {
                channel.send(x * x)
                println("send ${x * x}")
            }
        }

        val consumer = GlobalScope.launch {
//            val iterator = channel.iterator()
//            while (iterator.hasNext()) {
//                val next = iterator.next()
//                delay(2000)
//                println("receive $next")
//            }
            //eg2 这种也可以
            for (element in channel) {
                println("receive $element")
                delay(2000)
            }
        }

        joinAll(producer, consumer)
//        send 1
//        send 4
//        send 9
//        send 16
//        send 25
//        receive 1
//        receive 4
//        receive 9
//        receive 16
//        receive 25
    }

    @Test
    fun `test fast producer channel`() = runBlocking<Unit> {
        val receiverChannel: ReceiveChannel<Int> = GlobalScope.produce {
            repeat(100) {
                delay(1000)
                send(it)
            }
        }

        val consumer = GlobalScope.launch {
            for (i in receiverChannel) {
                println("received $i")
            }
        }
        consumer.join()
    }

    @Test
    fun `test fast consumer channel`() = runBlocking<Unit> {
        val sendChannel: SendChannel<Int> = GlobalScope.actor {
            while (true) {
                val element = receive()
                println("$element")
            }
        }
        val producer = GlobalScope.launch {
            for (i in 1..3) {
                sendChannel.send(i)
            }
        }
        producer.join()
    }

    @Test
    fun `test close channel`() = runBlocking<Unit> {
        val channel = Channel<Int>(3)
        val producer = GlobalScope.launch {
            List(3) {
                channel.send(it)
                println("send $it")
            }
            channel.close()
            println(
                """close channel.
                | - CloseForSend:${channel.isClosedForSend}
                | - CloseForReceive:${channel.isClosedForReceive}
            """.trimMargin()
            )
        }
        val consumer = GlobalScope.launch {
            for (element in channel) {
                println("receive=$element")
                delay(1000)
            }
            println(
                """After Consuming.
                | - CloseForSend:${channel.isClosedForSend}
                | - CloseForReceive:${channel.isClosedForReceive}
                
            """.trimMargin()
            )
        }
        joinAll(producer, consumer)
        //关闭最好由主导方来执行
    }

    @Test
    fun `test broadcast`() = runBlocking<Unit> {

        val channel  = Channel<Int>()
        val broadcastChannel = channel.broadcast(3)
//        val broadcastChannel = BroadcastChannel<Int>(Channel.BUFFERED)
        val producer = GlobalScope.launch {
            List(3) {
                delay(100)
                broadcastChannel.send(it)
            }
            broadcastChannel.close()
        }

        List(3) { index ->
            GlobalScope.launch {
                val receiveChannel = broadcastChannel.openSubscription()
                for (i in receiveChannel) {
                    println("#[$index] ---received:${i}")
                }
            }
        }

        producer.join()
    }

}